package com.nl.pulsardemo.obj.consumer;

import com.nl.pulsardemo.obj.bean.Person;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;

import java.util.concurrent.TimeUnit;

/**
 * @author : ZhaoY
 * @description :
 * @date : 2021-01-22 16:56
 **/
@Slf4j
public class ConsumerDemo {
    //Pulsar集群中broker的serviceUrl
    private static String brokerServiceUrl = "pulsar://10.1.8.68:6650,10.1.8.64:6650,10.1.8.69:6650";
    //需要订阅的topic name
    private static String topicName = "persistent://public/default/my-zyp";
    //订阅名
    private static String subscriptionName = "my-sub";

    public static void main(String[] args) throws PulsarClientException {
        //构造Pulsar client
        PulsarClient client = PulsarClient.builder()
                .serviceUrl(brokerServiceUrl)
                .build();

        //创建consumer
        Consumer<Person> consumer = client.newConsumer(Schema.JSON(Person.class))
                .topic(topicName)
                .subscriptionName(subscriptionName)
                .subscriptionType(SubscriptionType.Exclusive)//指定消费模式，包含：Exclusive，Failover，Shared，Key_Shared。默认Exclusive模式
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)//指定从哪里开始消费还有Latest，valueof可选，默认Latest
                .negativeAckRedeliveryDelay(60, TimeUnit.SECONDS)//指定消费失败后延迟多久broker重新发送消息给consumer，默认60s
                .subscribe();

        //消费消息
        while (true) {
            Message<Person> message = consumer.receive();

            try {
                //获取存储的数据
                Person person =  message.getValue();
                //原始数据格式 byte[]
                String result=new String(message.getData());
                log.info("person number"+person.getNumber());
                log.info("receive message: " + new String(message.getData()));
                consumer.acknowledge(message);
            } catch (Exception e) {
                e.printStackTrace();
                consumer.negativeAcknowledge(message);
            }

        }
    }


}
